package com.hoe.tsb.fraft.client;

import com.alibaba.fastjson.JSON;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.hoe.tsb.Configuration;
import com.hoe.tsb.block.Block;
import com.hoe.tsb.block.BlockChain;
import com.hoe.tsb.block.DefaultBlockChain;
import com.hoe.tsb.fraft.CurrentNode;
import com.hoe.tsb.fraft.Node;
import com.hoe.tsb.fraft.client.role.RoleAgent;
import com.hoe.tsb.fraft.enums.RoleType;
import com.hoe.tsb.fraft.event.msg.FRaftMessage;
import com.hoe.tsb.fraft.event.msg.IFRaftMessage;
import com.hoe.tsb.fraft.exception.FRaftException;
import com.hoe.tsb.fraft.listener.FRaftClientListener;
import com.hoe.tsb.fraft.net.Exchanger;
import com.hoe.tsb.fraft.net.ExchangerFactory;
import com.hoe.tsb.fraft.net.MessageReceive;
import com.hoe.tsb.mmk.TraceData;
import com.hoe.tsb.ring.RingSigner;
import com.hoe.tsb.tool.SerialUtils;
import com.hoe.tsb.web.NodeProperties;
import com.hoe.tsb.web.bean.Trace;
import com.hoe.tsb.web.bean.TraceResponse;
import com.xiaoleilu.hutool.crypto.digest.DigestUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.tomcat.util.codec.binary.Base64;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * <p>
 * 抽象FRaft客户端，实现基本业务逻辑，如角色转换、心跳维护等
 * </p>
 *
 * @author hoe
 * @since 2022/11/25
 */
@Slf4j
public abstract class AbstractClient implements FRaftClient, MessageReceive, FRaftClientListener, ClientRoleBridge {

    /**
     * 数据交换器
     */
    private final Exchanger mExchanger;
    /**
     * 区块链客户端
     */
    private final BlockChain mBlockChain;
    /**
     * 数据缓存池
     */
    private final List<TraceData> mDataPool = new ArrayList<>();
    /**
     * 用来生成区块的单线程线程池
     */
    private final Executor mGenBlockExecutor = new ThreadPoolExecutor(1, 1,
            60L, TimeUnit.SECONDS, new SynchronousQueue<>(),
            new ThreadFactoryBuilder().setNameFormat("client-task-%d").build(),
            (Runnable r, ThreadPoolExecutor executor) -> { throw new FRaftException("已无空闲ClientWorker线程，请稍后再试"); });
    /**
     * 配置类
     */
    protected final Configuration mConfig;
    /**
     * 对等网络节点集合<br/>
     * 当只有一个节点启动时该集合为空<br/>
     * 每次新节点加入时只需要连接集合中的任意节点即可<br/>
     * 连接成功后被连节点会将该节点广播到所有其他对等节点
     */
    protected List<Node> peers;
    /**
     * 当前节点
     */
    protected CurrentNode mCurrentNode;
    /**
     * 客户端监听集合
     */
    private final List<FRaftClientListener> mFRaftClientListeners = new ArrayList<>();
//    /**
//     * 任期（节点刚启动时，默认任期是0，随后每轮选举会自增）
//     */
//    protected AtomicInteger term;
//    /**
//     * 在该任期内是否已经投票了
//     */
//    protected AtomicBoolean hasVoted;
    /**
     * 代理角色
     */
    protected RoleAgent delegate;
    /**
     * 环签名器
     */
    protected RingSigner mRingSigner;

    public AbstractClient(NodeProperties properties, Configuration config) {
        CurrentNode current = properties.getCurrent();
        config.setCurrent(new Configuration.Info(current.getIp(), current.getPort(), current.getPublicKey()));
        config.setRingPrivateKey(current.getPrivateKey());
        config.setLevelDbDataPath(current.getDbPath());
        
        mConfig = config;
        delegate = new RoleAgent(this, config);
        mExchanger = ExchangerFactory.getExchanger(mConfig);
        mExchanger.setReceiver(this);
        mBlockChain = new DefaultBlockChain(config);
        peers = properties.getPeers();
        mCurrentNode = properties.getCurrent();
        mRingSigner = new RingSigner(mCurrentNode.getPrivateKey(), genPublicKeyArray(), 0);
        /**
         * 启动区块生成线程<br/>
         * 如果不是Leader，mDataPool会一直是空的，不会产生支链
         */
        mGenBlockExecutor.execute(() -> {
            try {
                while (true) {
                    synchronized (mDataPool) {
                        try {
                            mDataPool.wait(mConfig.getBlockGenInterval());
                        } catch (InterruptedException e) {
                            //ignore ...
                        }

                        //避免出空块
                        if (mDataPool.size() == 0) {
                            log.debug("数据缓存池无数据，跳过出块");
                            continue;
                        }

                        //有可能是Leader时保存了数据，但后面降级为Follower，此时需要把数据转发给现任Leader

                        log.info("检测到数据缓存池有{}条数据，开始生成区块并广播...", mDataPool.size());
                        //生成区块并广播
                        Block block = mBlockChain.gen(mDataPool);
                        //TODO 遍历mDataPool依次进行异步通知
                        mBlockChain.saveBlock(block);
                        broadcast(block);
                        mDataPool.clear();
                    }
                }
            } catch (Exception e) {
                log.error("生成区块出错", e);
            }
        });
    }

    @Override
    public void addListener(FRaftClientListener callback) {
        mFRaftClientListeners.add(callback);
    }

    @Override
    public void onReceive(FRaftMessage fRaftMessage) {
        log.debug("onReceive收到消息：{}", JSON.toJSONString(fRaftMessage));
        onMessageHandle(fRaftMessage);
    }

    @Override
    public void onMessageHandle(FRaftMessage msg) {
        //由各个FRaft角色进行处理
        boolean handled = delegate.handleMessage(msg);

        if (!handled) {
            //由client本身处理数据
            handleMessage(msg);
        }

        //由监听者们处理
        try {
            mFRaftClientListeners.forEach(l -> l.onMessageHandle(msg));
        } catch (Exception e) {
            log.error("客户端回调消息收到事件错误", e);
        }
    }

    @Override
    public long getHeight() {
        return mBlockChain.getHeight();
    }

    @Override
    public Block getBlock(String hash) {
        return mBlockChain.getBlock(hash);
    }

    @Override
    public String getBlockHash(String traceHash) {
        return mBlockChain.getBlockHash(traceHash);
    }

    @Override
    public void markLeader(String ip, int port) {
        peers.forEach(p -> p.setLeader(p.getIp().equals(ip) && p.getPort() == port));
    }

    @Override
    public void sendToLeader(IFRaftMessage message) {
        List<Node> leader = peers.stream().filter(Node::isLeader).collect(Collectors.toList());
        if (CollectionUtils.isEmpty(leader)) {
            throw new FRaftException("暂无Leader，请等待选举成功后再试");
        }
        doSend(createFRaftMsg(message, leader));
    }

    @Override
    public void sendToSupporters(IFRaftMessage message) {
        List<Node> supporters = peers.stream().filter(n -> !n.isLeader()).collect(Collectors.toList());
        doSend(createFRaftMsg(message, supporters));
    }

    @Override
    public void broadcast(IFRaftMessage message) {
        FRaftMessage fRaftMsg = createFRaftMsg(message, peers);
        doSend(fRaftMsg);
    }

    @Override
    public void sendToSpecific(IFRaftMessage message, String ip, int port) {
        List<Node> specific = peers.stream().filter(n -> n.getIp().equals(ip) && n.getPort() == port).collect(Collectors.toList());
        doSend(createFRaftMsg(message, specific));
    }

    @Override
    public Node getLeader() {
        return peers.stream().filter(Node::isLeader).findFirst().orElse(mCurrentNode);
    }

    @Override
    public int getNodeCount() {
        //还有一个节点是自身
        return peers.size() + 1;
    }

    @Override
    public Object submit(Trace trace) {
        //计算Hash
        String hash = DigestUtil.sha256Hex(SerialUtils.serialize(trace));
        trace.setTraceHash(hash);

        //计算签名（注意，Hash需要先计算）
        byte[][] sign = mRingSigner.sign(SerialUtils.serialize(trace));
        String signBase64 = Base64.encodeBase64String(sign[0]);
        String randomKBase64 = Base64.encodeBase64String(sign[1]);

        if (delegate.getRoleType() == RoleType.LEADER) {
            log.info("收到溯源数据提交请求，当前就是Leader，保存到数据池...");
            addTrace(trace);
        } else {
            log.info("收到溯源数据提交请求，当前不是Leader，尝试转发给Leader...");
            sendToLeader(trace);
        }

        //默认实现同步提交
        TraceResponse response = new TraceResponse();
        response.setHash(hash);
        response.setSign(signBase64);
        response.setSignRandom(randomKBase64);
        return response;
    }

    @Override
    public Block getTopBlock() {
        return getBlock(BlockChain.LATEST_BLOCK_HASH_KEY);
    }

    /**
     * 生成公钥数组<br/>
     * 当前公钥默认放在首位
     * @return 包含所有节点公钥的数组
     */
    private String[] genPublicKeyArray() {
        int size = peers.size();
        String[] publicArr = new String[size + 1];
        publicArr[0] = mCurrentNode.getPublicKey();
        for (int idx = 1, len = publicArr.length; idx < len; idx++) {
            publicArr[idx] = peers.get(idx - 1).getPublicKey();
        }
        return publicArr;
    }

    private void addTrace(Trace trace) {
        synchronized (mDataPool) {
            mDataPool.add(trace);
            //超过最大门限数量直接出块，避免无谓地等待
            if (mDataPool.size() >= mConfig.getBlockGenThreshold()) {
                mDataPool.notifyAll();
            }
        }
    }

    /**
     * 处理同步数据
     * @param msg 收到的FRaft消息
     */
    private void handleMessage(FRaftMessage msg) {
        switch (msg.getType()) {
            /*
             * 普通消息-溯源
             */
            case TRACE:
                log.info("收到来自Follower转发的溯源数据，保存到数据池...");
                //目前只要是Leader直接存入pool，不存在会转发给非Leader节点的情况，故这里直接保存即可
                Trace trace = SerialUtils.deserializeIFRaftMsg(msg.getBody(), Trace.class);
                addTrace(trace);
                break;
            /*
             * 普通消息-区块
             */
            case BLOCK:
                log.info("收到来自Leader转发的区块数据，保存到LevelDB...");
                //同理，这里也肯定是Leader出块后向follower们广播，或者新节点加入主动sync时收到的
                Block block = SerialUtils.deserializeIFRaftMsg(msg.getBody(), Block.class);
                //TODO 签名验证，区块数据验证
                mBlockChain.saveBlock(block);
                break;
        }
    }

    /**
     * 创建FRaft的发送消息
     * @param msg 待发送的源数据
     * @param nodes 待接收数据的节点集合
     * @return 创建好的FRaft消息
     */
    private FRaftMessage createFRaftMsg(IFRaftMessage msg, List<Node> nodes) {
        RoleType roleType = delegate.getRoleType();
        int term = delegate.getTerm();
        byte[] body = SerialUtils.serializeIFRaftMsg(msg);
        return new FRaftMessage(msg.getType(), term, roleType, body, nodes);
    }

    /**
     * 实际执行发送
     * @param msg 待发送消息
     */
    private void doSend(FRaftMessage msg) {
        log.debug("doSend发送消息：{}", JSON.toJSONString(msg));
        mExchanger.send(msg);
    }

}
